package com.hulk.rpc.transport.server.rpc.codec;

import com.hulk.rpc.config.common.HulkConstants;
import com.hulk.rpc.serialize.Serializer;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
import lombok.extern.slf4j.Slf4j;

/**
 * @author xuxukang
 */
@Slf4j
public class RequestDecoder extends LengthFieldBasedFrameDecoder {
  private final Serializer serializer;

  public RequestDecoder(int maxFrameLength, Serializer serializer) {
    super(maxFrameLength, 0, HulkConstants.HEADER_FIELD_LENGTH, 0,
            HulkConstants.HEADER_FIELD_LENGTH);
    this.serializer = serializer;
  }

  @Override
  protected Object decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
    ByteBuf buffer = (ByteBuf) super.decode(ctx, in);

    if (buffer != null) {
      try {
        return serializer.readRequest(buffer);
      } finally {
        buffer.release();
      }
    }

    return null;
  }

  @Override
  public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    if (log.isErrorEnabled()) {
      log.error("Exception caught on " + ctx.channel(), cause);
    }

    ctx.channel().close();
  }
}
